package com.ice.client.proto;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang.exception.ExceptionUtils;
import org.springframework.stereotype.Component;

import com.ice.framework.common.exception.BusinessException;
import com.ice.remoting.NettyRemotingClient;
import com.ice.remoting.common.ThreadFactoryImpl;
import com.ice.remoting.common.ThreadUtils;
import com.ice.remoting.data.RemotingCommand;
import com.ice.remoting.exception.RemotingConnectException;
import com.ice.remoting.exception.RemotingSendRequestException;
import com.ice.remoting.exception.RemotingTimeoutException;
import com.ice.remoting.netty.InvokeCallback;

import io.netty.channel.Channel;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.MessageToByteEncoder;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
public class ProtobufClient extends NettyRemotingClient {

    private Channel channel = null;
    protected ScheduledExecutorService brokerHeartbeatExecutorService;

    public ProtobufClient(ProtoNettyTestConfig nettyClientConfig) {
        super(nettyClientConfig);
    }

    public void start() throws InterruptedException {
        Channel channel = doConnect(new InetSocketAddress("localhost", 4001));
        if (channel == null || !channel.isActive()) {
            throw new BusinessException("channel未注册或未激活");
        }
        this.channel = channel;
        this.registerProcessor(ProtoMsg.HeadType.SCANCODEREQ.getNumber(), new ProtoDefaultProcess(this), null);

        this.brokerHeartbeatExecutorService =
            ThreadUtils.newScheduledThreadPool(1, new ThreadFactoryImpl("BrokerControllerHeartbeatScheduledThread"));

        scheduleSendHeartbeat();
    }

    public void stop() {
        super.shutdown();
        if (channel != null) {
            channel.close();
        }
        this.brokerHeartbeatExecutorService.shutdown();
    }

    protected void scheduleSendHeartbeat() {
        this.brokerHeartbeatExecutorService.scheduleAtFixedRate(() -> {
            try {
                RemotingCommand remotingCommand = new RemotingCommand();
                remotingCommand.setOpaque(RemotingCommand.createNewRequestId());
                remotingCommand.setCommand((byte)1);
                ProtoMsg.Message.Builder builder = ProtoMsg.Message.newBuilder();
                builder.setType(ProtoMsg.HeadType.HERART);
                ProtoMsg.HeartBeatReq.Builder heartBeatReq = ProtoMsg.HeartBeatReq.newBuilder();
                heartBeatReq.setY(2024);
                heartBeatReq.setM(2);
                heartBeatReq.setD(10);
                heartBeatReq.setH(16);
                heartBeatReq.setMi(21);
                heartBeatReq.setS(20);
                builder.setHeartBeatReq(heartBeatReq);
                remotingCommand.setBody(builder.build().toByteArray());
                invokeOneway(channel, remotingCommand);
            } catch (RemotingSendRequestException e) {
                log.info("发送失败：{}", ExceptionUtils.getStackTrace(e));
            }
        }, 1000, 2000, TimeUnit.MILLISECONDS);
    }

    public Channel getChannel() {
        return channel;
    }

    @Override
    public ByteToMessageDecoder buildDecoder() {
        return new ProtobufMsgDecode();
    }

    @Override
    public MessageToByteEncoder buildEncoder() {
        return new ProtobufMsgEncode();
    }

    public RemotingCommand invokeSync(RemotingCommand request)
        throws RemotingSendRequestException, RemotingTimeoutException, InterruptedException {
        return super.invokeSync(this.channel, request);
    }

    public RemotingCommand invokeSync(RemotingCommand request, Long requestTimeoutMillis)
        throws RemotingSendRequestException, RemotingTimeoutException, InterruptedException {
        return super.invokeSync(this.channel, request, requestTimeoutMillis);
    }

    public void invokeAsync(RemotingCommand request, InvokeCallback invokeCallback) throws RemotingConnectException {
        super.invokeAsync(this.channel, request, invokeCallback);
    }

    public void invokeAsync(RemotingCommand request, InvokeCallback invokeCallback, Long requestTimeoutMillis)
        throws RemotingConnectException {
        super.invokeAsync(this.channel, request, invokeCallback, requestTimeoutMillis);
    }

    public CompletableFuture<RemotingCommand> invokeAsyncFuture(RemotingCommand request) {
        return super.invokeAsyncFuture(this.channel, request);
    }
}
